#include "reactor.h"
#include "thread.h"
#include"public.h"
#include"RpcHeader.pb.h"
#include"RpcProvider.h"
#include <string.h>
#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include <unistd.h>
#include<string>
#include<google/protobuf/descriptor.h>
#include<google/protobuf/service.h>
/* 
内存池 stl sgi 二级空间配置器   nginx
线程池
进程池
对象池
连接池
都是为了资源的服用
*/

Thread::Thread() {
    if (pthread_create(&m_tid, NULL, taskFunc, this)) {
        std::cout << "creat thread failed!";
    }
    // pipe(int* fd)
    if (-1 == socketpair(AF_UNIX, SOCK_STREAM, 0, m_channel)) {
        std::cout << strerror(errno);
    }
}


Thread::~Thread() {
    pthread_join(m_tid, NULL);
}


int Thread::GetSocketPairFirst() const {
    return m_channel[0];
}

int Thread::GetSocketPairSecond() const {
    return m_channel[1];
}

pthread_t Thread::GetTid() const {
    return m_tid;
}

// one loop per thread reactor是抽象的 muduo libevent对reactor的一种实现 是具体的
// java netty
void* Thread::taskFunc(void* arg) {
    Thread* thread = static_cast<Thread*>(arg);
    thread->m_reactor = new Reactor();
    thread->m_reactor->AddEventAndHander(thread->GetSocketPairSecond(), EV_READ | EV_PERSIST, Thread::SocketPairEventCallBack, arg);
    std::cout << thread->GetTid()<<std::endl;
    thread->m_reactor->Loop();
    return NULL;
}   

void Thread::SocketPairEventCallBack(int fd, short events, void* arg) {
    Thread* thread = static_cast<Thread*>(arg);
    int cfd = 0;
    if (read(fd, &cfd, 4) <= 0) {
        std::cout << strerror(errno)<<std::endl;
    }
    // client io事件
    thread->m_reactor->AddEventAndHander(cfd, EV_READ | EV_PERSIST, Thread::ClientIOEventCallBack, arg);
}

/*
header_size(4个字节) + header_str + args_str
std::string   insert和copy方法 
*/
void Thread::ClientIOEventCallBack(int fd, short events, void* arg) {
    Thread* thread = static_cast<Thread*>(arg);
    if(events & EV_READ){
    
    char buf[LARGE_BUF_LEN] = {0}; // wx 
    if(read(fd, buf, LARGE_BUF_LEN) <0)
    {
        std::cout<<fd<<"close"<<std::endl;
    }
    //
    std::string header_size_str(buf);
    char *result = NULL;
    for (int i = 0; i < sizeof(buf); ++i)
	{
		if (buf[i] == '\n')
		{
			result = buf + i ;
			break;
		}
	}
    std::string data;

    if(result != nullptr)
      data+=result;
    else 
    {
        std::cout<<" parse error!" <<std::endl;
    }
    // 从字符流中读取前4个字节的内容
    uint32_t header_size = 0;
    header_size_str.copy((char*)&header_size, 4, 0);

    // 根据header_size读取数据头的原始字符流，反序列化数据，得到rpc请求的详细信息
    std::string rpc_header_str = data.substr(0, header_size);
    rpc::RPC_HEADER rpcHeader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size;
    if (rpcHeader.ParseFromString(rpc_header_str))
    {
        // 数据头反序列化成功
        service_name = rpcHeader.servicename();
        method_name = rpcHeader.methodname();
        args_size = rpcHeader.argsize();
    }
    else
    {
        // 数据头反序列化失败
        std::cout << "rpc_header_str:" << rpc_header_str << " parse error!" << std::endl;
        return;
    }

    // 获取rpc方法参数的字符流数据
    std::string args_str = data.substr(header_size, args_size);

    // 打印调试信息
    std::cout << "============================================" << std::endl;
    std::cout << "header_size: " << header_size << std::endl; 
    std::cout << "rpc_header_str: " << rpc_header_str << std::endl; 
    std::cout << "service_name: " << service_name << std::endl; 
    std::cout << "method_name: " << method_name << std::endl; 
    std::cout << "args_str: " << args_str << std::endl; 
    std::cout << "============================================" << std::endl;

    // 获取service对象和method对象

    auto it = RpcProvider::GetServiceMap().find(service_name);
    if (it == RpcProvider::GetServiceMap().end())
    {
        std::cout << service_name << " is not exist!" << std::endl;
        return;
    }

    auto mit = it->second.m_methodMap.find(method_name);
    if (mit == it->second.m_methodMap.end())
    {
        std::cout << service_name << ":" << method_name << " is not exist!" << std::endl;
        return;
    }

    google::protobuf::Service *service = it->second.service; // 获取service对象  new UserService
    const google::protobuf::MethodDescriptor *method = mit->second; // 获取method对象  Login

    // 生成rpc方法调用的请求request和响应response参数
    google::protobuf::Message *request = service->GetRequestPrototype(method).New();
    if (!request->ParseFromString(args_str))
    {
        std::cout << "request parse error, content:" << args_str << std::endl;
        return;
    }
    google::protobuf::Message *response = service->GetResponsePrototype(method).New();

    // 给下面的method方法的调用，绑定一个Closure的回调函数
    google::protobuf::Closure *done = google::protobuf::NewCallback<int, 
                                                                    google::protobuf::Message*>
                                                                    (&Thread::SendRpcResponse, 
                                                                    fd, response);

    // 在框架上根据远端rpc请求，调用当前rpc节点上发布的方法
    // new UserService().Login(controller, request, response, done)
    service->CallMethod(method, nullptr, request, response, done);
    }
}
void  Thread::SendRpcResponse(int cfd,google::protobuf::Message *response)
{
     std::string response_str;
      if (response->SerializeToString(&response_str)) // response进行序列化
    {
        // 序列化成功后，通过网络把rpc方法执行的结果发送会rpc的调用方
        int numberOfSendByte = send(cfd, response_str.c_str(), response_str.size(), 0);
       if (-1 == numberOfSendByte) {
          std::cout << strerror(errno)<<std::endl;
       }
    }
    else
    {
        std::cout << "serialize response_str error!" << std::endl; 
    }
    close(cfd); // 模拟http的短链接服务，由rpcprovider主动断开连接
    std::cout << "close " <<cfd<< std::endl; 
}

// int main() {
//     Thread t;

//     sleep(1);
//     return 0;
// }